package com.aotain.jupiter.flink;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.flink.api.common.functions.MapFunction;

import com.aotain.common.CommonFunction;

public class SourceDataFilterMap implements MapFunction<String, String> {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 6137189732292460920L;


	static HashSet<String> IPSet;
	static HashSet<String> WhiteIPSet;
	final long timeInterval = 1000*60*15;  
	String dbJson ;
	public SourceDataFilterMap(final String filterIpSql,final String whiteIpSql,final String dbJson) {
		this.dbJson = dbJson;

		Runnable runnable = new Runnable() {  
			public void run() {  
				while (true) {  
					try {  
						IPSet = IPFilterUtil.getFilterIpSet(filterIpSql,dbJson);
						WhiteIPSet = IPFilterUtil.getWhiteIpSet(whiteIpSql, dbJson);
						System.out.println("WihteIPSet update over ! Set'size is " + IPSet.size());  
						Thread.sleep(timeInterval);  
					} catch (InterruptedException e) {  
						e.printStackTrace();  
					}  
				}  
			}  
		};  
		Thread thread = new Thread(runnable);  
		thread.start();  
	}

	@Override
	public String map(String value) throws Exception {
		// TODO Auto-generated method stub

		String dipL = "";
		String dip = "";
		String sipL = "";
		String sip = "";

		String[] splits = value.split("\\|",-1);
		if(splits.length>1) {
			String userAccount = splits[1];
			String ipRex = "([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}";   
			Pattern pattern = Pattern.compile(ipRex);   
			Matcher matcher = pattern.matcher(userAccount);  
			if(matcher.matches()) {

				if("1".equals(splits[22]) && "6".equals(splits[7]) && splits.length >43) {//http
					//if(splits.length ==43) {//http
					dipL = splits[9];
					sipL = splits[12];
				}else{//非Http
					dipL = splits[9];
					sipL = splits[12];
				}/*else {
					return;
				}*/
				dip = CommonFunction.int2ip(Long.parseLong(dipL));
				sip = CommonFunction.int2ip(Long.parseLong(sipL));
				if((IPSet == null || IPSet.contains(dip) || IPSet.contains(sip))&&!WhiteIPSet.contains(dip)) {
					
					return value;
				}
			}

		}
		return "";
	}

}
